package com.yanggu.flink.datastream_api.sink

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}


/**
 * 把flink的DataStream数据输出到redis中
 * https://bahir.apache.org/docs/flink/current/flink-streaming-redis/
 */
object RedisSinkDemo {

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment

    val conf = new FlinkJedisPoolConfig.Builder()
      .setHost("127.0.0.1")
      .setDatabase(0)
      .setPort(6379)
      .build()
    environment
      .fromCollection(List(("1", "1"), ("2", "2"), ("3", "3")))
      .addSink(new RedisSink[(String, String)](conf, new RedisExampleMapper))

    environment.execute("RedisSinkDemo Job")

  }

}

/**
 * @tparam T redis处理的数据类型
 */
class RedisExampleMapper extends RedisMapper[(String, String)] {

  /**
   * 指定使用redis的数据类型
   * 这里使用string, 可以使用set
   *
   * @return
   */
  override def getCommandDescription: RedisCommandDescription = {
    //HASH和SORTED_SET需要添加额外的key
    new RedisCommandDescription(RedisCommand.HSET, "test-hash")
  }

  //这里的获取的key就是redis中内层hash的key, 如果是string其他的类型, 直接就是redis的key
  override def getKeyFromData(data: (String, String)): String = data._1

  override def getValueFromData(data: (String, String)): String = data._2

}
